package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.annotations.NotPoFieldAnnotation;
import com.atguigu.gmall.realtime.common.GmallConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 操作 ClickHouse 的工具类
 *
 * @author lvbingbing
 * @date 2022-07-03 19:22
 */
@Slf4j
public class ClickhouseUtils {

    private ClickhouseUtils() {
    }

    /**
     * 获取 jdbc sink
     *
     * @param sql 要执行的sql
     * @param <E> 流中的元素
     * @return Jdbc SinkFunction
     */
    public static <E> SinkFunction<E> operateClickhouseJdbcSink(String sql) {
        // 1、填充占位符并获取JdbcStatementBuilder
        JdbcStatementBuilder<E> jdbcStatementBuilder = fillPlaceholdersAndGetJdbcStatementBuilder();
        // 2、设置 Jdbc 执行选项
        JdbcExecutionOptions jdbcExecutionOptions = new JdbcExecutionOptions.Builder()
                // 批处理数据大小(一次处理多少条数据，默认为5000条)
                .withBatchSize(5)
                // 批处理时间间隔(隔多久执行一次，单位为ms，默认为0，不设置时间间隔)
                .withBatchIntervalMs(1000)
                // 失败重试次数，默认为3次
                .withMaxRetries(3)
                .build();
        // 3、设置 Jdbc 连接选项
        JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                .withUrl(GmallConfig.CLICKHOUSE_URL)
                // .withUsername() 如果clickhouse设置了用户名密码，连接时需要指定
                // .withPassword()
                // 设置连接最大超时时间，默认为 60 秒。
                .withConnectionCheckTimeoutSeconds(60)
                .build();
        // 4、返回 JdbcSink
        return JdbcSink.sink(sql, jdbcStatementBuilder, jdbcExecutionOptions, jdbcConnectionOptions);
    }

    /**
     * 根据流中元素，填充占位符并返回JdbcStatementBuilder
     *
     * @param <E> 流中的元素
     * @return org.apache.flink.connector.jdbc.JdbcStatementBuilder
     */
    public static <E> JdbcStatementBuilder<E> fillPlaceholdersAndGetJdbcStatementBuilder() {
        return new JdbcStatementBuilder<E>() {
            @Override
            public void accept(PreparedStatement preparedStatement, E element) throws SQLException {
                // 1、获取当前对象中声明的属性
                Field[] fields = element.getClass().getDeclaredFields();
                int cnt = 0;
                int length = fields.length;
                // 2、遍历所有属性，填充占位符
                for (int i = 0; i < length; i++) {
                    Field field = fields[i];
                    field.setAccessible(true);
                    // 3、如果当前属性不是 PO 属性，即非数据库中存在的字段，跳过填充该占位符
                    NotPoFieldAnnotation annotation = field.getAnnotation(NotPoFieldAnnotation.class);
                    if (annotation != null) {
                        cnt++;
                        continue;
                    }
                    Object fieldValue = null;
                    try {
                        fieldValue = field.get(element);
                        preparedStatement.setObject(i + 1 - cnt, fieldValue);
                    } catch (Exception e) {
                        String s = JSON.toJSONString(element);
                        log.error("设置对象属性值出现异常！当前操作对象：{}, 当前操作属性：{}, 异常原因：{}", s, fieldValue, e);
                    }
                }
            }
        };
    }
}
